package com.sunsharing.common.mailbox.memory;

import com.sunsharing.common.Constant;
import com.sunsharing.common.Context;
import com.sunsharing.common.ContextCreate;
import com.sunsharing.common.URL;
import com.sunsharing.common.anno.Factory;
import com.sunsharing.common.mailbox.AbstractMailbox;
import com.sunsharing.common.mailbox.Mailbox;
import com.sunsharing.common.mailbox.QueueHandle;
import com.sunsharing.common.utils.StringUtils;
import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 *
 * User: criss
 * Date: 13-4-2
 * Time: 下午10:51
 * To change this template use File | Settings | File Templates.
 */
@Factory(key="mem")
public class MemoryMailBox extends AbstractMailbox implements Mailbox {

    URL url;
    boolean isThreadSafe;
    public MemoryMailBox(URL url,Boolean isThreadSafe)
    {
        this.url = url;
        this.isThreadSafe = isThreadSafe;
    }

    LinkedBlockingQueue<Context> queue = new LinkedBlockingQueue<Context>();
    @Override
    public void addQueue(Context context) {
        try
        {
            //logger.error("ADDD!!!!!!!!!");
            length++;
            queue.put(context);
        }catch (Exception e)
        {

        }
    }

    final static Logger logger  = Logger.getLogger(MemoryMailBox.class);

    @Override
    public Context deQueue() {
        try
        {
            Context context = queue.take();
            return context;
        }catch (Exception e)
        {
            logger.error("出队列失败",e);
            throw new RuntimeException("出队列失败");
        }
    }

    ExecutorService executorService = null;
    RunTack tack = null;
    @Override
    public void startListen() {
        tack = new RunTack() ;
        Thread tackThread = new Thread(tack);
        tackThread.start();
        if(isThreadSafe)
        {
            if(executorService!=null)
            {
                executorService.shutdown();
                executorService = null;
            }
            executorService = Executors.newCachedThreadPool();
            //executorService = Executors.newFixedThreadPool(20);
        }
    }

    public void stopListen()
    {
        if(tack!=null)
        {
         try
         {
            if(queue.isEmpty())
            {
                addQueue(ContextCreate.createMailBoxStopContext());
            }else
            {
                tack.stop = true;
            }
            tack = null;
             if(isThreadSafe)
             {
                if(executorService!=null)
                {
                    executorService.shutdown();
                    executorService = null;
                }
             }

         }catch (Exception e)
         {
             logger.error("",e);
         }
        }
    }

    class RunTack implements Runnable
    {
        final  Logger logger  = Logger.getLogger(RunTack.class);
        boolean stop = false;
        public void run()
        {
            logger.info("mailbox---------" + url + "----------开始监听请求");
            while(!stop)
            {
                try
                {
                    final Context contect = queue.take();
                    length--;
                    if(contect.getMsgType()== Constant.MSG_TYPE_STOP)
                    {
                        break;
                    }
                    addOne(contect);
                    //System.out.println(contect);
                    for(Iterator iter = list.iterator();iter.hasNext();)
                    {
                        final QueueHandle handle = (QueueHandle)iter.next();
                        if(isThreadSafe)
                        {
                            executorService.execute(new Runnable() {
                                @Override
                                public void run() {
                                    handle.run(contect);
                                }
                            });
                        }else
                        {

                            handle.run(contect);
                        }
                        //
                    }
                } catch (Exception e)
                {
                     logger.error("",e);
                }
            }
            logger.info("mailbox------" + url + "--------推出消息监听");
        }

    }



}


